package com.yanggu.market_analysis

import cn.hutool.core.date.DateUtil
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.util.Date
import scala.util.Random

/**
 * 以考察不分渠道的市场推广统计，这样得到的就是所有渠道推广的总量
 * 统计过去1h内访问数量, 每隔5s输出一次
 */
object AppMarketingStatistics {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    environment
      .addSource(new SimulatedEventSource)
      .filter(_.behavior != "UNINSTALL")
      .assignAscendingTimestamps(_.timestamp)
      //设置随机key, 进行hash散列, 充分使用分布式计算的能力
      .map(_ => Random.nextString(8))
      //按照key进行分组
      .keyBy(data => data)
      //设置滑动时间窗口, 窗口大小1h, 滑动步长5s
      .window(SlidingEventTimeWindows.of(Time.hours(1L), Time.seconds(5L)))
      //当Watermark大于等于windowEnd时, 执行该函数。具有窗口内的全部数据
      .process(new MarketingCountFunction)
      //按照key进行hash散列之后需要重新进行聚合, 按照endTime重新进行聚合
      .keyBy(_.endTime)
      .process(new MarketingCountKeyedProcessFunction)
      .print("result")

    environment.execute("AppMarketingByChannel Job")
  }

}

case class MarketingViewCount(startTime: Long, endTime: Long, count: Long)

class MarketingCountFunction extends ProcessWindowFunction[String, MarketingViewCount, String, TimeWindow] {

  override def process(key: String, context: Context, elements: Iterable[String], out: Collector[MarketingViewCount]): Unit = {
    out.collect(MarketingViewCount(context.window.getStart, context.window.getEnd, elements.size))
  }

}

class MarketingCountKeyedProcessFunction extends KeyedProcessFunction[Long, MarketingViewCount, String] {

  private lazy val countValueState = getRuntimeContext.getState(
    new ValueStateDescriptor[Long]("countValueState", classOf[Long]))

  //分组内的每一个元素都会执行该方法
  override def processElement(value: MarketingViewCount,
                              ctx: KeyedProcessFunction[Long, MarketingViewCount, String]#Context,
                              out: Collector[String]): Unit = {
    countValueState.update(countValueState.value() + 1)
    //注册定时器, 在窗口结束时间 + 1ms后执行
    ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey + 1L)
  }

  override def onTimer(timestamp: Long,
                       ctx: KeyedProcessFunction[Long, MarketingViewCount, String]#OnTimerContext,
                       out: Collector[String]): Unit = {
    val result = s"系统当前时间: ${DateUtil.formatDateTime(new Date())}, " +
      s"开始时间: ${DateUtil.formatDateTime(new Date(ctx.getCurrentKey - 1 * 60 * 60 * 1000))}" +
      s" - 结束时间: ${DateUtil.formatDateTime(new Date(ctx.getCurrentKey))}, 过去1小时内, 访问数量: ${countValueState.value()}"
    out.collect(result)
    //清空状态
    countValueState.clear()
  }

}